package cn.doitedu.demo1;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 *    实时监控所有在线用户的行为，按如下规则(rule-001)进行计算：
 *        用户发生 浏览（“商务休闲”）商品的行为次数： >= 5次
 *        当他发生 添加购物车（“商务休闲”）行为时，给他发送促销信息
 *        
 *  {"user_id":1,"event_id":"page_load","action_time":12378573468,"properties":{"url":"/abdc/dde/a.html?sdosdg","title":"商务休闲速溶咖啡喝了不困提神减肥" }}
 *  {"user_id":1,"event_id":"page_load","action_time":12378573468,"properties":{"url":"/attc/xxe/b.html?sdosdg","title":"仰望U8男人都想拥有比亚迪最高端汽车" }}      
 *  {"user_id":1,"event_id":"add_cart","action_time":12378573468,"properties":{"item_id":"item_001","title":"仰望U8男人都想拥有比亚迪最高端汽车" }}      
 **/
public class MainEntry {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
        
        // 构建kafka source来读取数据
        KafkaSource<String> source = KafkaSource
                .<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("user-action-log")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doitedu-ww")
                .setClientIdPrefix("doitedu-cc")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "log-source");
        
        // 将source读取到的json数据，转换成javaBean数据
        DataStream<UserAction> mapped = stream.map(json -> JSON.parseObject(json, UserAction.class));

        // 按用户分区
        KeyedStream<UserAction, Long> keyedStream = mapped.keyBy(UserAction::getUser_id);
        
        // 核心计算逻辑
        keyedStream.process(new KeyedProcessFunction<Long, UserAction, String>() {

            ValueState<Integer> aEventCntState;
            final JSONObject jsonObject = new JSONObject();

            @Override
            public void open(Configuration parameters) throws Exception {

                RuntimeContext runtimeContext = getRuntimeContext();
                aEventCntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("A_EVENT_CNT", Integer.class));

                jsonObject.put("rule_id","rule-001");

            }

            @Override
            public void processElement(UserAction userAction, KeyedProcessFunction<Long, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                // 用户发生 浏览（“商务休闲”）商品的行为次数： >= 5次
                String eventId = userAction.getEvent_id();
                HashMap<String, String> eventProperties = userAction.getProperties();

                // 统计A行为的次数
                if("page_load".equals(eventId) && eventProperties.get("title").contains("商务休闲")){
                    Integer oldCount = aEventCntState.value();
                    oldCount = oldCount == null? 0 : oldCount;

                    aEventCntState.update(oldCount+1);
                }


                // 当他发生 添加购物车（“商务休闲”）行为时，给他发送促销信息
                if("add_cart".equals(eventId)){

                    Integer currentACount = aEventCntState.value();
                    if( currentACount!=null &&  currentACount >= 5 ){

                        // 输出规则命中的结果
                        jsonObject.put("user_id",userAction.getUser_id());
                        jsonObject.put("hit_time",userAction.getAction_time());


                        out.collect(jsonObject.toJSONString());

                    }

                }

            }
        }).print();



        env.execute();
        
        
        


    }
}
